package com.wx.redis.mq;

import cn.hutool.core.lang.UUID;
import com.google.gson.Gson;
import jakarta.annotation.PostConstruct;
import jakarta.annotation.Resource;
import lombok.extern.slf4j.Slf4j;
import org.redisson.api.RBucket;
import org.redisson.api.RTopic;
import org.redisson.api.RedissonClient;
import org.slf4j.MDC;

import java.util.concurrent.TimeUnit;


/**
 * @author weixin 2024/11/20 15:26
 */
@Slf4j
public abstract class AbstractMqListener {

    @Resource
    private RedissonClient redissonClient;

    @PostConstruct
    public void execute() {
        Gson gson = new Gson();

        String topic = getQueue();
        RTopic rTopic = redissonClient.getTopic(topic);
        rTopic.addListenerAsync(String.class, (charSequence, body) -> {
            String trackId = UUID.fastUUID().toString(true);
            MDC.put("trackId",trackId);

            MqMessageBo object = check(body, gson, topic);
            if (object == null) return;

            try {
                log.info("收到消息。mqId:{},topic:{},param:{}",object.getMqId(), topic, gson.toJson(object.getBody()));

                doJob(object.getBody());

            } catch (Exception e) {
                log.error("消费失败。topic:{},param:{}", topic, gson.toJson(object.getBody()),e);
            }
        });
    }

    private MqMessageBo check(String body, Gson gson, String topic) {
        MqMessageBo object = gson.fromJson(body, MqMessageBo.class);
        if (object == null) {
            log.error("mq参数解析异常。topic:{},body:{}", topic, body);
            return null;
        }
        // 避免同一个服务多节点重复消费
        RBucket<Object> bucket = redissonClient.getBucket(topic + "_" + object.getMqId());
        if(bucket.get() != null){
            log.info("消息mqId:{} 已被消费，无需重复消费",object.getMqId());
            return null;
        }
        // 设置1分钟
        bucket.set("aa",1L, TimeUnit.MINUTES);
        return object;
    }


    /**
     * 获取订阅的队列
     */
    protected abstract String getQueue();

    /**
     * 业务处理
     */
    protected abstract void doJob(String param);

}
